use std::collections::{BTreeMap, HashMap};
use std::io::stdout;
use std::marker::PhantomData;
use std::ops::{Deref, DerefMut};
use std::pin::Pin;
use std::sync::{Arc, Barrier, Mutex};
use std::task::ready;
use std::time::Duration;
use std::{mem, ptr};

use bytes::Bytes;
use mini_redis::server::run;
use serde_json::ser::State;
use tokio::sync::broadcast::Sender;
use tokio::sync::Notify;
use tokio::time::Instant;

use crate::utils::multi_cond::SimpleCountDownLatch;
use crate::{safe_pin, thread_safe_ptr};

/// mini_redis的存储中心，维护一个hashMap
///
/// 设计想法：
/// {Db{kv(键值对),exp(过期表),补充一个pub/sub},notify(过期通知),expireTask(过期任务)}
///
/// DB:
/// 1.需要共享(Arc)
/// 2.释放时需要判断是否结“束过定时期任务“(Drop)

/// Entry:
/// 1.id(唯一ID，可能性能比较好（类比hashCode）)
/// 2.content(内容)
/// 3.expire(过期时间点)
#[derive(Debug)]
struct Entry {
    unique_id: u64,             //要支持标准hashcode位长度
    content: Bytes,             //使用bytes避免重复拷贝
    expire_at: Option<Instant>, //instant仅用于定位过期时间点
}

/// Store:
/// 1.entries(标准的kv表)
/// 2.expirations(过期时间表)
/// 3.pub_sub:(订阅队列)
/// 4.next_id(下一位ID)
/// 5.shout_down(定之标记位)
#[derive(Debug)]
struct Store {
    entries: HashMap<String, Entry>,
    expirations: BTreeMap<(Instant, u64), String>,
    pub_sub: HashMap<String, Sender<Bytes>>, //A multi-producer, multi-consumer broadcast queue. Each sent value is seen by all consumers.
    next_id: u64,
    shut_down: bool,
}

impl Store {
    //获取下一个要过期的时间点
    fn next_expiration(&self) -> Option<&Instant> {
        self.expirations.keys().next().map(|(instant, id)| instant)
    }
}

#[derive(Debug)]
struct SharedState {
    state: Mutex<Store>,     //使用std::Mutex保证插入一定是全局安全的
    exp_task_caller: Notify, //维系一个唤起器用于控制，过期定时任务的执行
}

impl SharedState {
    //清理所有的过期的键值，返回下一个要过期的时间点
    ///此处的Instant可能会因为set导致移除，一定要拷贝处理
    fn purge_expired_keys(&self) -> Option<Instant> {
        let mut state = self.state.lock().unwrap();
        if state.shut_down {
            return None;
        }
        let state = state.deref_mut(); //在锁的保证下直接获取可变引用避免重复强转
        let now = Instant::now();
        while let Some((&(instant, id), key)) = state.expirations.iter().next() {
            if instant > now {
                return Some(instant);
            }
            state.entries.remove(key);
            state.expirations.remove(&(instant, id));
        }

        None
    }

    fn is_shut_down(&self) -> bool {
        self.state.lock().unwrap().shut_down
    }
}

/// Db:
/// 标准的Db，支持多线程下的安全访问
#[derive(Debug)]
pub struct Db {
    db: Arc<SharedState>,
}

impl Drop for Db {
    fn drop(&mut self) {
        //当且仅当db被主线程持和后台过期任务同时绑定获取时
        if Arc::strong_count(&self.db) == 2 {
            let mut guard = self.db.state.lock().unwrap();
            guard.shut_down = true;
            drop(guard);
            self.db.exp_task_caller.notify_one();
        }
    }
}

impl Db {
    //维护一个过期任务，精度在毫秒级
    async fn purge_expired_key_task(db: Arc<SharedState>) {
        while !db.is_shut_down() {
            if let Some(instant) = db.purge_expired_keys() {
                //睡眠到下一个过期时间or被唤起（说明有更早的过期时间被插入）
                tokio::select! {
                    _= tokio::time::sleep_until(instant)=>{}
                    _= db.exp_task_caller.notified()=>{}
                }
            } else {
                //没有键值对，让出时间片，等待唤起工作
                db.exp_task_caller.notified().await;
            }
        }
    }
    pub fn new() -> Self {
        let db = Arc::new(SharedState {
            state: Mutex::new(Store {
                entries: HashMap::new(),
                pub_sub: HashMap::new(),
                expirations: BTreeMap::new(),
                next_id: 0,
                shut_down: false,
            }),
            exp_task_caller: Notify::new(),
        });
        // tokio::spawn(Db::purge_expired_key_task(db.clone()));
        Self { db }
    }

    ///通过key值获取
    pub fn get(&self, key: &str) -> Option<Bytes> {
        let store = self.db.state.lock().unwrap();
        store.entries.get(key).map(|e| e.content.clone())
    }
    pub fn ttl(&self, key: &str) -> Option<Duration> {
        let store = self.db.state.lock().unwrap();
        if let Some(Entry { expire_at, .. }) = store.entries.get(key) {
            if let &Some(instant) = expire_at {
                return Some(instant - Instant::now());
            }
        }
        None
    }

    ///查看所有现存的keys
    pub fn keys(&self, key_pattern: &str) -> Vec<String> {
        const SEARCH_ALL: &str = "*";
        let store = self.db.state.lock().unwrap();
        store
            .entries
            .keys()
            .filter_map(|k| {
                if k.contains(key_pattern) || SEARCH_ALL.eq(key_pattern) {
                    return Some(k.clone());
                }
                None
            })
            .collect::<Vec<String>>()
    }

    pub fn del(&self, key: &str) -> Option<Bytes> {
        let mut result = None;
        let mut store = self.db.state.lock().unwrap();
        if let Some(Entry {
            unique_id,
            content,
            expire_at: Some(instant),
        }) = store.entries.remove(key)
        {
            store.expirations.remove(&(instant, unique_id));
            result = Some(content);
        }
        drop(store);
        self.db.exp_task_caller.notify_one();
        result
    }

    pub fn set(&self, key: &str, val: Bytes, expire: Option<Duration>) {
        let mut store = self.db.state.lock().unwrap();
        let this_id = store.next_id;
        store.next_id += 1;
        let mut should_notify = false;
        //如果当前进入的过期时间小于下一个现有的时间，需要唤起过期任务进行工作
        let now_expire_at = expire.map(|duration| {
            let now_expire_at = Instant::now() + duration;
            should_notify = store
                .next_expiration()
                .map(|&next_expire_at| next_expire_at > now_expire_at)
                .unwrap_or(true);
            //先插入过期map中（lock保证整个流程不会被打断）
            store
                .expirations
                .insert((now_expire_at, this_id), key.to_string());
            now_expire_at
        });

        //再插入hashMap中
        let old_kv = store.entries.insert(
            key.to_string(),
            Entry {
                unique_id: this_id,
                content: val,
                expire_at: now_expire_at,
            },
        );
        //由于expire_map_key值包含时间和唯一id，当old_kv被新的取代之后，过期表可能无法被清理
        if let Some(entry) = old_kv {
            if let Some(when) = entry.expire_at {
                store.expirations.remove(&(when, entry.unique_id));
            }
        }
        //结束锁周期
        drop(store);
        if should_notify {
            self.db.exp_task_caller.notify_one();
        }
    }
    // 订阅
    pub fn subscribe(&self, key: &str) -> tokio::sync::broadcast::Receiver<Bytes> {
        use std::collections::hash_map::Entry;
        let mut store = self.db.state.lock().unwrap();
        match store.pub_sub.entry(key.to_string()) {
            Entry::Occupied(o) => o.get().subscribe(),
            Entry::Vacant(v) => {
                let (sd, rc) = tokio::sync::broadcast::channel(1024);
                v.insert(sd);
                rc
            }
        }
    }
    // 发布
    pub fn publish(&self, key: &str, val: Bytes) -> usize {
        let store = self.db.state.lock().unwrap();
        store
            .pub_sub
            .get(key)
            .map(|e| e.send(val).unwrap_or(0))
            .unwrap_or(0)
    }
}
//##################################################################################################//
///过期时效测试在毫秒级
#[tokio::test]
async fn redis_db_test() {
    let mut db = Db::new();
    db.set("a", Bytes::from("shit"), Some(Duration::from_millis(100)));
    tokio::time::sleep(Duration::from_millis(10)).await;
    let option = db.ttl("a");
    println!("{:?}", option);
    // db.set(
    //     "shit",
    //     Bytes::from("shit"),
    //     Some(Duration::from_millis(390)),
    // );
    //
    // let option = db.get("shit");
    // println!("{:?}", option);
    // tokio::time::sleep(Duration::from_millis(400)).await;
    // let option = db.del("shit");
    // println!("{:?}", option);
}
///发布订阅设置
#[tokio::test]
async fn redis_pub_sub() {
    let mut db = Db::new();
    let mut db = db;
    thread_safe_ptr!(db);
    let key = "ps";
    let mut receiver = db.subscribe(key);
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(2)).await;
        db.publish(key, "everything is ok".into());
    });
    let msg = receiver.recv().await.unwrap_or("no msg".into());
    println!("{:?}", msg);
}

#[test]
fn redis_multi_test() {
    let mut db = Db::new();
    let runtime = tokio::runtime::Runtime::new().unwrap();
    thread_safe_ptr!(db);
    runtime.spawn(Db::purge_expired_key_task(db.db.clone()));
    let mut pool = super::super::utils::thread_pool::SimplePool::new(16);
    let len = 10_0000;
    thread_safe_ptr! {
            let latch = SimpleCountDownLatch::new_with_lock(len);
    }
    for i in 0..len {
        pool.commit(move || {
            db.set(
                format!("{}", i).as_str(),
                Bytes::from("this is a check"),
                None,
            );
            latch.count_down();
        });
    }
    latch.wait();

    let vec = db.keys("*");
    println!(
        "{}, counter{:?}",
        vec.len(),
        *super::super::utils::multi_cond::COUNTER_PTR
    );
}
